package com.wei.violet.utils

import org.apache.commons.lang.StringUtils
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.Row
object BaseDataKpi {

  def apply(d: Row)={



      val requestmode = d.getAs[Int]("requestmode")
      val processnode = d.getAs[Int]("processnode")
      val iseffective = d.getAs[Int]("iseffective")
      val isbilling = d.getAs[Int]("isbilling")
      val isbid = d.getAs[Int]("isbid")
      val iswin = d.getAs[Int]("iswin")
      val adorderid = d.getAs[Int]("adorderid")
      val winprice = d.getAs[Double]("winprice")
      val adpayment = d.getAs[Double]("adpayment")

      var InitialRequest = if (requestmode == 1 && processnode >= 1) 1 else 0
      var EffectiveRequest = if (requestmode == 1 && processnode >= 2) 1 else 0
      var ADRequests = if (requestmode == 1 && processnode == 3) 1 else 0
      var BiddingCount = if (iseffective == 1 && isbilling == 1 && isbid == 1 && adorderid != 0) 1 else 0
      var ADShow = if (requestmode == 2 && iseffective == 1) 1 else 0
      var ADHits = if (requestmode == 3 && iseffective == 1) 1 else 0
      var BiddingSuccess = 0
      var DSPADExpense = 0d
      var DSPADCosting = 0d
      if (iseffective == 1 && isbilling == 1 && iswin == 1) {
        BiddingSuccess = 1
        DSPADExpense = winprice / 1000d
        DSPADCosting = adpayment / 1000d
      }


      (List(InitialRequest, EffectiveRequest, ADRequests, BiddingCount, BiddingSuccess, ADShow, ADHits, DSPADExpense, DSPADCosting))



  }

}
